package org.example.flink;

import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.ParquetRowInputFormat;
import org.apache.flink.formats.parquet.ParquetTableSource;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;
import org.apache.parquet.schema.Type;

import java.util.ArrayList;
import java.util.List;

public class BatchDemo {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);
        tEnv.registerFunction("mutilFun", new MutilFun());
//        List<Type> typeList = new ArrayList<>();
//        typeList.add(new MessageType("name"));
//        typeList.add(new MessageType("type"));
//        MessageType messageType = new MessageType("", typeList);
//        ParquetTableSource tableSource = ParquetTableSource.builder()
//                .path("hdfs://ha:9000/data/person.parquet")
//                .forParquetSchema(messageType)
//                .build();
//        Table table1 = tEnv.fromTableSource(tableSource);
//        tEnv.toDataSet(table1, String.class).groupBy("")
//                .reduceGroup((GroupReduceFunction<String, String>) (values, out) -> {
//
//        });
//        DataSource<Row> rowDataSource = env.readFile(new ParquetRowInputFormat(new Path("/data"), messageType), "/data");
//        DataSource<Person> personDataSource = env.fromElements(Person.class, new Person("1"), new Person("2,3"));
//        DataSource<String> dataSource = env.fromElements("1", "2,2,3,42,13", "3");

//        tEnv.createTemporaryView("test_table", personDataSource);
//        Table table = tEnv.sqlQuery("select * from test_table");
//        table.printSchema();
//        DataSet<Person> result = tEnv.toDataSet(table, Person.class);
//        result.print();
    }
}
